package com.echo.code.thread.mode.listener;

import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;

/**
 * Dispatcher Created by baiyap on 2018/9/24.
 */
public class Dispatcher {

  private final ExecutorService executorService;
  private final EventExceptionHandler exceptionHandler;

  public Dispatcher(ExecutorService executorService, EventExceptionHandler exceptionHandler) {
    this.executorService = executorService;
    this.exceptionHandler = exceptionHandler;
  }

  public void dispatch(Registry registry, Object event, String topic) {
    // 根据topic获取所有的Subscriber列表
    ConcurrentLinkedQueue<Subscriber> subscribers = registry.scanSubscriber(topic);
    if (Objects.isNull(subscribers)) {
      if (Objects.nonNull(exceptionHandler)) {
        exceptionHandler.handle(new IllegalArgumentException("The topic " + topic + " not bind yet"));
      }
      return;
    }

    // 遍历所有的方法, 并且通过反射的方式进行方法调用
    subscribers.stream().filter(subscriber -> !subscriber.isDisable()).filter(subscriber -> {
      Method subscribeMethod = subscriber.getSubscribeMethod();
      Class<?> clazz = subscribeMethod.getParameterTypes()[0];
      return (clazz.isAssignableFrom(event.getClass()));
    }).forEach(subscriber -> realInvokeSubscribe(subscriber, event));
  }

  private void realInvokeSubscribe(Subscriber subscriber, Object event) {
    Method subscribeMethod = subscriber.getSubscribeMethod();
    Object subscribeObject = subscriber.getSubscribeObject();
    executorService.execute(() -> {
      try {
        subscribeMethod.invoke(subscribeObject, event);
      } catch (Exception e) {
        if (Objects.nonNull(exceptionHandler)) {
          exceptionHandler.handle(e);
        }
      }
    });
  }

  public void close() {
    executorService.shutdown();
  }
}
